package fun.carliels.doraemon.common.io.monitor;

import org.apache.commons.io.monitor.FileAlterationListener;
import org.apache.commons.io.monitor.FileAlterationMonitor;
import org.apache.commons.io.monitor.FileAlterationObserver;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.util.concurrent.*;
import java.util.function.Consumer;

/**
 * 文件监听器
 * <p>监听指定目录下的文件，且等待文件就绪后进行消费</p>
 * @author 残血拉二胡
 */
public class FileMonitor implements Runnable{

    private static final Logger log  = LoggerFactory.getLogger(FileMonitor.class);

    /**
     * 监听目录
     */
    private String dir;

    /**
     * 监听间隔
     */
    private long interval;

    /**
     * 结果处理函数
     */
    private Consumer<String> consumer;

    /**
     * 运行状态
     */
    private volatile boolean running;

    /**
     * 等待队列消费线程
     */
    private Thread thread;

    /**
     * 线程工厂
     */
    private ThreadFactory threadFactory;

    /**
     * 目录监听器
     */
    private FileAlterationMonitor monitor;

    /**
     * 未准备好的文件
     */
    private CopyOnWriteArraySet<String> unreadyFileSet;

    /**
     * 文件等待队列
     */
    private DelayQueue<WaitingFile> waitingQueue;

    /**
     * 等待时间
     */
    private long waitingExpire;

    public FileMonitor(String dir, long interval, Consumer<String> consumer) {
        this.dir = dir;
        this.interval = interval < 100 ? 100 : interval;
        this.consumer = consumer;
        this.thread = null;
        this.running = false;
        // 创建等待队列
        this.waitingQueue = new DelayQueue<>();
        // 设置等待时间, 扫描时间加上3s
        this.waitingExpire = this.interval + 3000;
        // 创建未就绪文件标记集
        this.unreadyFileSet = new CopyOnWriteArraySet<>();
    }

    /**
     * 启动监听
     */
    public synchronized void start() throws Exception {
        if (this.running) {
            throw new IllegalStateException("Monitor is already running");
        } else {
            this.running = true;
            if (this.threadFactory != null) {
                this.thread = this.threadFactory.newThread(this);
            } else {
                this.thread = new Thread(this);
            }
            this.thread.start();
        }
    }

    /**
     * 停止监听
     * @throws Exception
     */
    public synchronized void stop() throws Exception {
        this.stop(this.interval);
    }

    /**
     * 停止监听
     * @throws Exception
     */
    public synchronized void stop(long stopInterval) throws Exception {
        if (!this.running) {
            throw new IllegalStateException("File Monitor is not running");
        } else {
            this.running = false;

            try {
                if (this.monitor != null) {
                    this.monitor.stop();
                }
                this.thread.interrupt();
                this.thread.join(stopInterval);
            } catch (InterruptedException var5) {
                Thread.currentThread().interrupt();
            }
        }
    }

    @Override
    public void run() {
        try {
            // 创建目录监听器
            startMonitoring();
            // 消费文件
            consumeReadyFile(this.consumer);
        } catch (Exception e) {
            log.error("FileMonitor running exception.", e);
        }
    }

    /**
     * 创建监听器
     */
    private void startMonitoring() throws Exception {
        this.monitor = new FileAlterationMonitor(this.interval);
        FileAlterationObserver observer = new FileAlterationObserver(new File(this.dir));
        observer.addListener(new FileAlterationListener() {
            @Override
            public void onStart(FileAlterationObserver fileAlterationObserver) {}
            @Override
            public void onDirectoryCreate(File file) {}
            @Override
            public void onDirectoryChange(File file) {}
            @Override
            public void onDirectoryDelete(File file) {}

            @Override
            public void onFileCreate(File file) {
                putToWaitingQueue(file.getAbsolutePath());
            }
            @Override
            public void onFileChange(File file) {
                markUnreadyFile(file.getAbsolutePath());
            }

            @Override
            public void onFileDelete(File file) {}
            @Override
            public void onStop(FileAlterationObserver fileAlterationObserver) {}
        });
        this.monitor.addObserver(observer);
        this.monitor.start();
    }

    /**
     * 把文件放置到等待队列
     */
    private void putToWaitingQueue(String filePath) {
        System.out.println(String.format("%d - waiting in: %s", System.currentTimeMillis(), filePath));
        waitingQueue.put(new WaitingFile(filePath, waitingExpire));
    }

    /**
     * 记录未准备好的文件
     * @param filePath
     */
    private void markUnreadyFile(String filePath) {
        System.out.println(String.format("%d - mark unready: %s", System.currentTimeMillis(), filePath));
        if (!unreadyFileSet.contains(filePath)) {
            unreadyFileSet.add(filePath);
        }
    }

    /**
     * 消费需要保存的文件
     */
    private void consumeReadyFile(Consumer<String> consumer) throws InterruptedException {
        while (true) {
            System.out.println("poll from waiting queue");
            if (!this.running) {
                System.out.println("stopping...");
                break;
            }
            WaitingFile countDownFile = waitingQueue.poll(100, TimeUnit.MILLISECONDS);
            if (countDownFile != null){
                // 文件还有被标记的情况,重新放置到队列中
                if (unreadyFileSet.contains(countDownFile.getFilePath())) {
                    putToWaitingQueue(countDownFile.getFilePath());
                    unreadyFileSet.remove(countDownFile.getFilePath());
                } else {
                    // 没有被标记的文件，直接推送事件，进行写一部动作
                    consumer.accept(countDownFile.getFilePath());
                }
            }
        }
    }

    public String getDir() {
        return dir;
    }

    public void setDir(String dir) {
        this.dir = dir;
    }

    public long getInterval() {
        return interval;
    }

    public void setInterval(long interval) {
        this.interval = interval;
    }

    public Consumer<String> getConsumer() {
        return consumer;
    }

    public void setConsumer(Consumer<String> consumer) {
        this.consumer = consumer;
    }

    public ThreadFactory getThreadFactory() {
        return threadFactory;
    }

    public void setThreadFactory(ThreadFactory threadFactory) {
        this.threadFactory = threadFactory;
    }

    public boolean isRunning() {
        return running;
    }

    /**
     * 等待文件对象
     */
    private class WaitingFile implements Delayed {

        private String filePath;
        private long expire;

        public WaitingFile(String filePath, long expire) {
            this.filePath = filePath;
            this.expire = expire + System.currentTimeMillis();
        }

        @Override
        public long getDelay(TimeUnit unit) {
            long diffTime = expire - System.currentTimeMillis();
            return unit.convert(diffTime,TimeUnit.MILLISECONDS);
        }

        @Override
        public int compareTo(Delayed o) {
            return (int)(this.expire - ((WaitingFile) o).getExpire());
        }

        public String getFilePath() {
            return filePath;
        }

        public void setFilePath(String filePath) {
            this.filePath = filePath;
        }

        public long getExpire() {
            return expire;
        }

        public void setExpire(long expire) {
            this.expire = expire;
        }
    }
}
